package cn.ac.iie.ulss.indexer.metastore;

import cn.ac.iie.ulss.indexer.runenvs.Constants;
import cn.ac.iie.ulss.indexer.runenvs.GlobalParas;
import cn.ac.iie.ulss.indexer.worker.IndexControler;
import cn.ac.iie.ulss.indexer.worker.TableChange;
import com.taobao.metamorphosis.Message;
import com.taobao.metamorphosis.client.MessageSessionFactory;
import com.taobao.metamorphosis.client.MetaClientConfig;
import com.taobao.metamorphosis.client.MetaMessageSessionFactory;
import com.taobao.metamorphosis.client.consumer.ConsumerConfig;
import com.taobao.metamorphosis.client.consumer.MessageConsumer;
import com.taobao.metamorphosis.client.consumer.MessageListener;
import com.taobao.metamorphosis.exception.MetaClientException;
import com.taobao.metamorphosis.utils.ZkUtils;
import java.text.SimpleDateFormat;
import java.util.Iterator;
import java.util.concurrent.Executor;
import org.apache.hadoop.hive.metastore.msg.MSGFactory.DDLMsg;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

public class GetMetastoreMessage implements Runnable {

    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    static Logger log = Logger.getLogger(GetMetastoreMessage.class.getName());
    int attempSize = 2;

    static {
        PropertyConfigurator.configure("log4j.properties");
    }

    @Override
    public void run() {
        log.info("getting the message from the metastore");
        String mszkurl = GlobalParas.mszkUrl;
        String zkUrl = mszkurl.split("\\|")[0];
        String topic = mszkurl.split("\\|")[1];
        String ccgroup = mszkurl.split("\\|")[2];
        MetaClientConfig metaClientConfig = new MetaClientConfig();
        final ZkUtils.ZKConfig zkConfig = new ZkUtils.ZKConfig();
        zkConfig.zkConnect = zkUrl;
        metaClientConfig.setZkConfig(zkConfig);
        final MessageSessionFactory sessionFactory;
        try {
            sessionFactory = new MetaMessageSessionFactory(metaClientConfig);
            ConsumerConfig cc = new ConsumerConfig("idx_" + GlobalParas.hostName + "_" + ccgroup);
            cc.setFetchRunnerCount(1);
            final MessageConsumer consumer = sessionFactory.createConsumer(cc);
            log.info("init the consumer ok,begin receive data from the metastore ");
            consumer.subscribe(topic, 5 * 1024 * 1024, new MessageListener() {
                @Override
                public void recieveMessages(Message message) throws InterruptedException {
                    DDLMsg msg = new DDLMsg();
                    String data = new String(message.getData());
                    msg = DDLMsg.fromJson(data);
                    int eventId = (int) msg.getEvent_id();
                    String db_name = null;
                    String table_name = null;
                    String indexName = null;
                    String column_name = null;
                    String column_type = null;
                    Long version = -1L;
                    TableChange tg = new TableChange();
                    boolean tableChange = true;
                    Iterator itor = GlobalParas.id2Createindex.keySet().iterator();
                    switch (eventId) {
                        case 1001: // create new db                 
                            db_name = (String) msg.getMsg_data().get("db_name");
                            log.info("the message from metastore is 1001 to createDb:" + db_name);
                            tg.createDB(db_name);
                            break;
                        case 1004: // drop db                           
                            db_name = (String) msg.getMsg_data().get("db_name");
                            log.info("the message from metastore is 1004 to deleteDb:" + db_name);
                            tg.deleteDB(db_name);
                            break;
                        case 1101: // create new table                         
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            log.info("the message from metastore is 1101 to createTable:" + db_name + "." + table_name);
                            tg.createTable(db_name, table_name);
                            break;

                        case 1102: // change table name

                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            String old_table_name = (String) msg.getMsg_data().get("old_table_name");
                            log.info("the message from metastore is 1102 " + db_name + "." + table_name + "," + old_table_name);
                            while (itor.hasNext()) {
                                IndexControler c = GlobalParas.id2Createindex.get((Long) itor.next());
                                if (c.dbName.equals(db_name) && c.tbName.equals(old_table_name)) {
                                    c.tableChangeFlag.getAndSet(Constants.TABLE_CHAGING_TIME);
                                }
                            }
                            while (true) {
                                tableChange = true;
                                Iterator itor1 = GlobalParas.id2Createindex.keySet().iterator();
                                while (itor1.hasNext()) {
                                    IndexControler c = GlobalParas.id2Createindex.get((Long) itor1.next());
                                    if (c.dbName.equals(db_name) && c.tbName.equals(old_table_name)) {
                                        if (c.tableChangeFlag.get() == Constants.TABLE_CHAGED_TIME) {

                                        } else {
                                            tableChange = false;
                                            break;
                                        }
                                    }
                                }
                                if (tableChange == true) {
                                    break;
                                }
                            }
                            tg.changeTableName(db_name, table_name, old_table_name);
                            break;

                        case 1105: // change the rule
                            log.info("the message from metastore for 1105!!! , begin to change the transmit rule");
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            version = Long.parseLong((String) msg.getMsg_data().get("version"));
                            break;

                        case 1201: // delete the index
                            log.info("the message from metastore for 1201!!!");
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            column_name = (String) msg.getMsg_data().get("column_name");
                            column_type = (String) msg.getMsg_data().get("column_type");
                            while (itor.hasNext()) {
                                IndexControler c = GlobalParas.id2Createindex.get((Long) itor.next());
                                if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                    c.tableChangeFlag.getAndSet(Constants.TABLE_CHAGING_TIME);
                                }
                            }
                            while (true) {
                                tableChange = true;
                                Iterator itor1 = GlobalParas.id2Createindex.keySet().iterator();
                                while (itor1.hasNext()) {
                                    IndexControler c = GlobalParas.id2Createindex.get((Long) itor1.next());
                                    if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                        if (c.tableChangeFlag.get() == Constants.TABLE_CHAGED_TIME) {

                                        } else {
                                            tableChange = false;
                                            break;
                                        }
                                    }
                                }
                                if (tableChange == true) {
                                    break;
                                }
                            }
                            tg.columnDelete(db_name, table_name, column_name, column_type);
                            break;

                        case 1202: //add col                          
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            column_name = (String) msg.getMsg_data().get("column_name");
                            column_type = (String) msg.getMsg_data().get("column_type");
                            log.info("the message from metastore for 1202 to add column:" + db_name + "." + table_name + "(" + column_name + "," + column_type + ")");
                            while (itor.hasNext()) {
                                IndexControler c = GlobalParas.id2Createindex.get((Long) itor.next());
                                if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                    c.tableChangeFlag.getAndSet(Constants.TABLE_CHAGING_TIME);
                                }
                            }
                            while (true) {
                                tableChange = true;
                                Iterator itor1 = GlobalParas.id2Createindex.keySet().iterator();
                                while (itor1.hasNext()) {
                                    IndexControler c = GlobalParas.id2Createindex.get((Long) itor1.next());
                                    if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                        if (c.tableChangeFlag.get() == Constants.TABLE_CHAGED_TIME) {

                                        } else {
                                            tableChange = false;
                                            break;
                                        }

                                    }
                                }
                                if (tableChange == true) {
                                    break;
                                }
                            }
                            log.info("the thread to write data wait,now begin to add column");
                            tg.columnAdd(db_name, table_name, column_name, column_type);
                            break;

                        case 1203: //change col name                        
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            column_name = (String) msg.getMsg_data().get("column_name");
                            String old_column_name = (String) msg.getMsg_data().get("old_column_name");
                            log.info("the message from metastore is 1203 for change the column name:" + old_column_name + "->" + column_name);
                            while (itor.hasNext()) {
                                IndexControler c = GlobalParas.id2Createindex.get((Long) itor.next());
                                if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                    c.tableChangeFlag.getAndSet(Constants.TABLE_CHAGING_TIME);
                                }
                            }
                            while (true) {
                                tableChange = true;
                                Iterator itor1 = GlobalParas.id2Createindex.keySet().iterator();
                                while (itor1.hasNext()) {
                                    IndexControler c = GlobalParas.id2Createindex.get((Long) itor1.next());
                                    if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                        if (c.tableChangeFlag.get() == Constants.TABLE_CHAGED_TIME) {

                                        } else {
                                            tableChange = false;
                                            break;
                                        }
                                    }
                                }
                                if (tableChange == true) {
                                    break;
                                }
                            }
                            tg.changeColumnName(db_name, table_name, column_name, old_column_name);
                            break;
                        case 1207: //drop table                            
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            log.info("the message from metastore is 1207 to drop the table:" + db_name + "." + table_name);
                            while (itor.hasNext()) {
                                IndexControler c = GlobalParas.id2Createindex.get((Long) itor.next());
                                if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                    c.tableChangeFlag.getAndSet(Constants.TABLE_CHAGING_TIME);
                                }
                            }
                            while (true) {
                                tableChange = true;
                                Iterator itor1 = GlobalParas.id2Createindex.keySet().iterator();
                                while (itor1.hasNext()) {
                                    IndexControler c = GlobalParas.id2Createindex.get((Long) itor1.next());
                                    if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                        if (c.tableChangeFlag.get() == Constants.TABLE_CHAGED_TIME) {

                                        } else {
                                            tableChange = false;
                                            break;
                                        }
                                    }
                                }
                                if (tableChange == true) {
                                    break;
                                }
                            }
                            tg.dropTable(db_name, table_name);
                            break;
                        case 1401: //create index
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            indexName = (String) msg.getMsg_data().get("index_name");
                            log.info("the message from metastore for 1401 to get the index:" + db_name + "." + table_name + "(" + indexName + ")");
                            while (itor.hasNext()) {
                                IndexControler c = GlobalParas.id2Createindex.get((Long) itor.next());
                                if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                    c.tableChangeFlag.getAndSet(Constants.TABLE_CHAGING_TIME);
                                }
                            }
                            while (true) {
                                Iterator itor1 = GlobalParas.id2Createindex.keySet().iterator();
                                tableChange = true;
                                while (itor1.hasNext()) {
                                    IndexControler c = GlobalParas.id2Createindex.get((Long) itor1.next());
                                    if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                        if (c.tableChangeFlag.get() == Constants.TABLE_CHAGED_TIME) {

                                        } else {
                                            tableChange = false;
                                            break;
                                        }
                                    }
                                }
                                if (tableChange == true) {
                                    break;
                                }
                            }
                            log.info("the thread to write data wait,now begin to get the index!");
                            tg.getIndex(db_name, table_name, indexName);
                            break;
                        case 1404: //delete index
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            indexName = (String) msg.getMsg_data().get("index_name");
                            log.info("the message from metastore is 1404 to delete the index:" + db_name + "." + table_name + "(" + indexName + ")");
                            while (itor.hasNext()) {
                                IndexControler c = GlobalParas.id2Createindex.get((Long) itor.next());
                                if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                    c.tableChangeFlag.getAndSet(Constants.TABLE_CHAGING_TIME);
                                }
                            }
                            while (true) {
                                Iterator itor1 = GlobalParas.id2Createindex.keySet().iterator();
                                tableChange = true;
                                while (itor1.hasNext()) {
                                    IndexControler c = GlobalParas.id2Createindex.get((Long) itor1.next());
                                    if (c.dbName.equals(db_name) && c.tbName.equals(table_name)) {
                                        if (c.tableChangeFlag.get() == Constants.TABLE_CHAGED_TIME) {

                                        } else {
                                            tableChange = false;
                                            break;
                                        }
                                    }
                                }
                                if (tableChange == true) {
                                    break;
                                }
                            }
                            log.info("the thread to write data wait,now begin to get the index!");
                            tg.deleteIndex(db_name, table_name, indexName);
                            break;
                        case 4002:     //truncate table
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            long f_id = Long.parseLong(msg.getMsg_data().get("f_id").toString());
                            log.info("the message from metastore is 4002 to delete the file:"+db_name+"."+table_name+" "+f_id);
                            IndexControler c1 = GlobalParas.id2Createindex.get(f_id);
                            if (c1 != null && c1.dbName.equals(db_name) && c1.tbName.equals(table_name)) {                           
                                GlobalParas.id2Createindex.remove(f_id);
                            }
                            log.info("the file " + db_name + "." + table_name + " "+f_id+" has been deleted!");
                            break;
                        case 1307: //change the file status
                            db_name = (String) msg.getMsg_data().get("db_name");
                            table_name = (String) msg.getMsg_data().get("table_name");
                            long file_id = Long.parseLong(msg.getMsg_data().get("f_id").toString());
                            String status = (String) msg.getMsg_data().get("new_status");
                            log.info("the message from metastore is 1307 for change the file " + file_id + "'s status to: " + status);
                            IndexControler c = GlobalParas.id2Createindex.get(file_id);
                            if (c != null && c.dbName.equals(db_name) && c.tbName.equals(table_name)&&"4".equals(status)) {
                                c.isOver = true;
                                GlobalParas.truncateFileId.add(file_id);                               
                                log.info("the file's " + file_id + " status has been changed");
                            }
                            break;
                        default:
                            log.info("Event " + eventId + " is useless");
                            break;
                    }
                }
                @Override
                public Executor getExecutor() {
                    return null;
                }
            });
            consumer.completeSubscribe();
        } catch (MetaClientException ex) {
            log.error(ex, ex);
        }
    }
}
